package org.budo.dubbo.protocol.async.repository.redis;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import org.budo.dubbo.protocol.async.BudoAsyncDubboProtocol;
import org.budo.dubbo.protocol.async.repository.AbstractAsyncRepository;
import org.budo.dubbo.protocol.async.repository.AsyncRepository;
import org.budo.dubbo.protocol.async.repository.BudoAsyncInvocation;
import org.budo.support.lang.util.StringUtil;

import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.rpc.Exporter;
import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;
import com.alibaba.dubbo.rpc.RpcInvocation;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPubSub;

/**
 * @author limingwei
 */
@Slf4j
@Getter
@Setter
public class RedisAsyncRepository extends AbstractAsyncRepository {
    private JedisPool jedisPool;

    @Override
    protected Integer _doPopMessage() {
        String[] channels = this.channels();

        if (null == channels || channels.length < 1) {
            log.error("#55 channels=" + channels + ", this=" + this);
            return 0;
        }

        JedisPubSub jedisPubSub = new JedisPubSub() {
            public void onMessage(String channel, String message) {
                Invoker invoker = RedisAsyncRepository.this.getInvoker(channel);

                BudoAsyncInvocation budoAsyncInvocation = RedisAsyncRepository.this.deserializeAsyncInvocation(message.getBytes(), invoker);

                RpcInvocation invocation = BudoAsyncInvocation.toDubboInvocation(budoAsyncInvocation);
                invoker.invoke(invocation); // 执行方法
            }
        };

        Jedis jedis = this.getJedisPool().getResource();
        jedis.subscribe(jedisPubSub, channels);
        return -1;
    }

    @Override
    public Integer popMessageAndInvoke(Invoker invoker) {
        throw new RuntimeException();
    }

    @Override
    public void sendMessageWhenInvoke(Invocation invocation) {
        Class<?> interfaceType = invocation.getInvoker().getInterface();
        String destinationName = invocation.getInvoker().getUrl().getParameter(AsyncRepository.DESTINATION_NAME);
        if (StringUtil.isEmpty(destinationName)) {
            destinationName = "budo-async-" + interfaceType.getName();
        }

        Jedis jedis = this.getJedisPool().getResource();

        BudoAsyncInvocation budoAsyncInvocation = BudoAsyncInvocation.fromDubboInvocation(invocation);
        byte[] requestBody = this.serializeAsyncInvocation(budoAsyncInvocation, invocation.getInvoker());

        jedis.publish(destinationName, new String(requestBody));
        jedis.close();
    }

    protected Invoker getInvoker(String channel) {
        BudoAsyncDubboProtocol dubboProtocol = BudoAsyncDubboProtocol.getInstance();

        if (null == dubboProtocol || null == dubboProtocol.getExporterMap() || dubboProtocol.getExporterMap().isEmpty()) {
            if (log.isDebugEnabled()) {
                log.debug("#154 dubboPopMessage error, protocol=" + dubboProtocol + ", this=" + this);
            }

            return null;
        }

        Collection<Exporter<?>> dubboExporters = dubboProtocol.getExporterMap().values();
        for (Exporter<?> dubboExporter : dubboExporters) { // 每一个
            Invoker<?> invoker = dubboExporter.getInvoker();
            String destinationName = invoker.getUrl().getParameter(DESTINATION_NAME);

            if (StringUtil.equals(destinationName, channel)) {
                return invoker;
            }
        }

        log.error("#105 channel=" + channel + ", invoker is null");
        return null;
    }

    private String[] channels() {
        BudoAsyncDubboProtocol dubboProtocol = BudoAsyncDubboProtocol.getInstance();

        if (null == dubboProtocol || null == dubboProtocol.getExporterMap() || dubboProtocol.getExporterMap().isEmpty()) {
            if (log.isDebugEnabled()) {
                log.debug("#154 dubboPopMessage error, protocol=" + dubboProtocol + ", this=" + this);
            }

            return null;
        }

        List<String> destinations = new ArrayList<String>();

        Collection<Exporter<?>> dubboExporters = dubboProtocol.getExporterMap().values();
        for (Exporter<?> dubboExporter : dubboExporters) { // 每一个
            Invoker<?> invoker = dubboExporter.getInvoker();
            URL url = invoker.getUrl();
            String asyncRepositoryBeanName = url.getParameter(ASYNC_REPOSITORY_BEAN_NAME);

            if (null == asyncRepositoryBeanName || asyncRepositoryBeanName.isEmpty()) {
                log.warn("#104 asyncRepositoryBeanName is null, invoker=" + invoker);
                continue;
            }

            if (!asyncRepositoryBeanName.equals(this.getBeanName())) { // 只处理指定当前Bean的
                if (log.isDebugEnabled()) {
                    log.debug("#175 asyncRepositoryBeanName=" + asyncRepositoryBeanName + ", this=" + this);
                }

                continue;
            }

            String destinationName = invoker.getUrl().getParameter(DESTINATION_NAME);
            destinations.add(destinationName);
        }

        return destinations.toArray(new String[destinations.size()]);
    }
}